package com.deepglint.transform;

import com.deepglint.beans.SensorReading;
import org.apache.commons.math3.analysis.function.StepFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-15 23:41
 */
public class TransformTest_RichFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换成SensorReading类型
        DataStream<SensorReading> streamMap = inputStream.map((MapFunction<String, SensorReading>) value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // richFunction
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = streamMap.map(new MyMapper());

        map.print();

        env.execute();
    }
}

class MyMapper extends RichMapFunction<SensorReading, Tuple2<String, Integer>> {

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化工作，一般是定义状态或者连接数据库
        System.out.println("open");
    }

    @Override
    public void close() throws Exception {
        // 一般是关闭连接或清空状态操作
        System.out.println("close");
    }

    @Override
    public Tuple2<String, Integer> map(SensorReading value) throws Exception {
        return new Tuple2<>(value.getId(), getRuntimeContext().getIndexOfThisSubtask());
    }
}